package org.hong.monkey.rpc

import java.io.File
import java.nio.channels.ReadableByteChannel

import org.hong.monkey.MonkeyConf
import org.hong.monkey.util.{RpcUtils, Utils}

import scala.concurrent.Future

/**
 */

private[monkey] object RpcEnv {
  private def getRpcEnvFactory(conf: MonkeyConf): RpcEnvFactory = {
    val rpcEnvNames = Map(
      "akka" -> "org.hong.monkey.rpc.akka.AkkaRpcEnvFactory",
      "netty" -> "org.hong.monkey.rpc.netty.NettyRpcEnvFactory"
    )
    val rpcEnvName = conf.get("monkey.rpc", "netty")
    val rpcEnvFactoryClassName =  rpcEnvNames.getOrElse(rpcEnvName.toLowerCase, rpcEnvName)
    Utils.classForName(rpcEnvFactoryClassName).newInstance().asInstanceOf[RpcEnvFactory]
  }

  def create(name:String, host:String, port:Int, conf:MonkeyConf, clientMode:Boolean = false): RpcEnv = {
    val config = RpcEnvConfig(conf, name, host, port, clientMode)
    getRpcEnvFactory(conf).create(config)
  }
}

private[monkey] abstract class RpcEnv(conf: MonkeyConf) {

  private[monkey] val defaultLookupTimeout = RpcUtils.lookupRpcTimeout(conf)

  private[rpc] def endpointRef(endpoint: RpcEndpoint): RpcEndpointRef

  def address: RpcAddress

  def setupEndpoint(name:String, endpoint: RpcEndpoint): RpcEndpointRef

  def asyncSetupEndpointRefByURI(uri: String): Future[RpcEndpointRef]

  def setupEndpointRefByURI(uri: String): RpcEndpointRef = {
    defaultLookupTimeout.awaitResult(asyncSetupEndpointRefByURI(uri))
  }

  def setupEndpointRef(systemName: String, address: RpcAddress, endpointName:String):RpcEndpointRef = {
    setupEndpointRefByURI(uriOf(systemName, address, endpointName))
  }

  def stop(endpoint: RpcEndpointRef): Unit

  def shutdown(): Unit

  def awaitTermination(): Unit

  def uriOf(systemName: String, address: RpcAddress, endpointName:String): String

  def deserialize[T](deserializationAction: () => T): T

  def fileServer: RpcEnvFileServer

  def openChannel(uri: String): ReadableByteChannel

}

private[monkey] trait RpcEnvFileServer {

  def addFile(file: File): String

  def addJar(file: File): String

}

private[monkey] case class RpcEnvConfig(
  conf: MonkeyConf,
  name:String,
  host:String,
  port: Int,
  clientMode: Boolean)